/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.data;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.Comparator;
import java.util.Iterator;
import java.util.TreeSet;

import taskgraph.ports.InputPort;
import taskgraph.ports.OutputPort;
import taskgraph.tasks.PrimitiveTask;


/**
 * Sorts objects in ascending or descending order.
 * 
 * @author Armando Blancas
 * @see taskgraph.ports.InputPort
 * @see taskgraph.ports.OutputPort
 * @param <E> The type of elements sorted by this task.
 */
public class Sort<E> extends PrimitiveTask {

	private InputPort<E> input;
	private OutputPort<E> output;
	private Comparator<? super E> comparator;
	
	/**
	 * This constructor allows creating instances as beans.
	 */
    public Sort() {
    	// do nothing
    }

	/**
     * Creates an instance that sorts in natural ordering.
     * 
     * @param input The input port.
     * @param output The output port.
     */
    public Sort(final InputPort<E> input, final OutputPort<E> output) {
    	setInput(input);
    	setOutput(output);
    }

	/**
     * Creates an instance that sorts using a comparator.
     * 
     * <p>For sorting in descending natural order, use the standard
     * comparator {@code Collections.reverseOrder()}.
     * 
     * @param input The input port.
     * @param output The output port.
	 * @param comparator The comparator used for sorting.
     */
    public Sort(final InputPort<E> input, 
    		    final OutputPort<E> output,
    		    final Comparator<? super E> comparator) {
    	setInput(input);
    	setOutput(output);
    	setComparator(comparator);
    }

    /**
	 * Gets the comparator used for sorting.
	 * 
	 * @return the comparator
	 */
	public Comparator<? super E> getComparator() {
		return comparator;
	}

    /**
     * Gets the input port.
     * 
	 * @return The input port.
	 */
	public InputPort<E> getInput() {
		return input;
	}
	
    /**
     * Gets the output port.
     * 
	 * @return The output port.
	 */
	public OutputPort<E> getOutput() {
		return output;
	}

	/**
     * Performs the work of this task.
     * 
	 * Reads references to element objects from the input port, sorts them, 
	 * and writes the sorted sequence to the output port. Sorting is done 
	 * incrementally using a variable-size data structure to allocate only 
	 * the required memory.
	 *<pre>
	 *while not EOF
	 *    read an object reference from input port
	 *    store it in a sorted collection
	 *close input port
	 *while not end of collection
	 *    get the next element
	 *    write it to the output port
	 *close output port
	 *</pre>
	 */
	@Override
    public void run() {
    	try {
    		TreeSet<E> set = new TreeSet<E>(comparator);
    		while (!Thread.currentThread().isInterrupted()) {
    			E element = null;
    			try {
    				element = input.read();
    			} catch (EOFException e) {
    				break;
    			}
    			set.add(element);
            }
    		for (Iterator<E> itor = set.iterator(); itor.hasNext(); ) {
    			output.write(itor.next());
    		}
        } catch (InterruptedIOException ie) {
        	// interruption request
        } catch (IOException e) {
        	// pipe synch error
            log.error("Sort#run()", e);
		} finally {
			input.close();
			output.close();
        }
    }

	/**
	 * Sets the comparator used for sorting.
	 * 
	 * @param comparator The comparator to set
	 */
	public void setComparator(final Comparator<? super E> comparator) {
    	if (comparator == null)
    		throw new IllegalArgumentException("comparator == null");
		this.comparator = comparator;
	}

	/**
	 * Sets the input port.
	 * 
	 * @param input The input port.
	 */
    public void setInput(final InputPort<E> input) {
    	if (input == null)
    		throw new IllegalArgumentException("input == null");
 		this.input = input;
	}

	/**
	 * Sets the output port.
	 * 
	 * @param output The output port.
	 */
    public void setOutput(final OutputPort<E> output) {
    	if (output == null)
    		throw new IllegalArgumentException("output == null");
 		this.output = output;
	}

}
